﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Configuration;
using System.Linq;
using log4net;
using Reactor.Environment;
using Reactor.Exceptions;
using Reactor.Messages.Events.Endpoint;
using Reactor.Messaging.ServiceBus.Internals;
using Samurai.Wakizashi;

namespace Reactor.Messaging.ServiceBus
{
    public sealed class DefaultBus : DisposableBase, IServiceBus
    {
        #region Fields

        private static readonly ILog Log = LogManager.GetLogger(typeof(DefaultBus));
        private readonly INmsGateway _nmsGateway;
        private int _defaultNumberOfConsumingSessions;
        private TaskDistributor _taskDistributor;
        private readonly ConcurrentQueue<KeyValuePair<IMessagingChannel, List<Type>>> _queuedSubscriptions = new ConcurrentQueue<KeyValuePair<IMessagingChannel, List<Type>>>();
        private readonly object _msgCountSync = new object();
        private ulong _sendCount;
        private ulong _receiveCount;

        #endregion

        #region Constructor

        /// <summary>
        /// Initializes a new instance of the <see cref="DefaultBus"/> class.
        /// </summary>
        /// <param name="nmsGateway">The JMS manager.</param>
        /// <param name="busConfiguration">The bus configuration.</param>
        public DefaultBus(INmsGateway nmsGateway, BusConfiguration busConfiguration)
        {
            if (nmsGateway == null) throw new ArgumentNullException("nmsGateway");
            if (busConfiguration == null) throw new ArgumentNullException("busConfiguration");

            _nmsGateway = nmsGateway;

            // Ensure configuration
            if(busConfiguration.DestinationRegistry == null)
                throw new InvalidOperationException("No IDestinationRegistry instance configured.");

            if(busConfiguration.MessageHandlerRegistry == null)
                throw new InvalidOperationException("No IMessageHandlerRegistry instance configured.");

            Configuration = busConfiguration;

            // Need to know when the handler registry finds message handlers
            Configuration.MessageHandlerRegistry.HandlersRegistered += MessageHandlerRegistry_HandlersRegistered;

            // MessagingChannel registry must have a named subscriber to represent subscriptions by this bus instance.
            busConfiguration.DestinationRegistry.InstanceSubscriber.Name = busConfiguration.CurrentEndpointInputChannel.DestinationName;

            // Determine default number of consuming sessions
            DetermineDefaultNumberOfConsumingSessions();
        }

        #endregion

        #region Implementation of IServiceBus

        #region Events

        /// <summary>
        /// Occurs when a message is sent.
        /// </summary>
        public event EventHandler<MessageEventArgs> MessageSent;

        /// <summary>
        /// Occurs when a message is received.
        /// </summary>
        public event EventHandler<MessageEventArgs> MessageReceived;

        /// <summary>
        /// Occurs when the service bus run status has changed.
        /// </summary>
        public event EventHandler<RunStatusChangedEventArgs> RunStatusChanged;

        /// <summary>
        /// Occurs when subscriptions are updated. That is, whenever a
        /// subscription is created or terminated.
        /// </summary>
        public event EventHandler<SubscriptionUpdateEventArgs> SubscriptionsUpdated;

        #endregion

        /// <summary>
        /// Gets or sets the configuration instance this bus is to use.
        /// </summary>
        /// <value>The configuration.</value>
        public BusConfiguration Configuration { get; set; }

        /// <summary>
        /// Gets or sets a value indicating whether this instance is running.
        /// </summary>
        /// <value>
        /// 	<c>true</c> if this instance is running; otherwise, <c>false</c>.
        /// </value>
        public bool IsRunning { get; private set; }

        /// <summary>
        /// Starts the bus by opening a connection to the underlying broker(s). If connection fails, a <seealso cref="Reactor.Messaging.Exceptions.ConnectionException"/> is thrown.
        /// </summary>
        public void Start()
        {
            try
            {
                if(Log.IsDebugEnabled) Log.Debug("Attempting to start service bus instance...");

                // We must have an input queue for all primary endpoints. Attempt to resolve that input queue for the current endpoint.
                if(Configuration.CurrentEndpointInputChannel == null)
                {
                    if (Context.GridContext.CurrentReactorService == null)
                        throw new InvalidOperationException("No current endpoint input queue defined and no Context.CurrentReactorService found. Either set the CurrentEndpointInputChannel property or a current service instance.");

                    Configuration.CurrentEndpointInputChannel = Context.GridContext.CurrentReactorService.PrimaryInputChannel; 
                }

                // Need to know when the handler registry finds message handlers
                Configuration.MessageHandlerRegistry.HandlersRegistered += MessageHandlerRegistry_HandlersRegistered;

                // Start up our transport gateway
                _nmsGateway.Startup();

                // See if we need to flush primary input queue and do so now if needed
                if (Configuration.FlushPrimaryInputQueueOnStartup)
                    _nmsGateway.DeleteDestination(Configuration.CurrentEndpointInputChannel);

                _taskDistributor = new TaskDistributor(_nmsGateway, _defaultNumberOfConsumingSessions);
                _taskDistributor.MessageSent += TaskDistributor_MessageSent;
                _taskDistributor.MessageReceived += TaskDistributor_MessageReceived;
                _taskDistributor.Start();

                Context.ServiceLocator.RegisterInstance<TaskDistributor>(_taskDistributor);

                SubmitQueuedSubscriptionRequestsToTaskDistributor();

                IsRunning = true;
                if (Log.IsDebugEnabled) Log.Debug("Service bus instance is running.");
                if (RunStatusChanged != null) RunStatusChanged(this, new RunStatusChangedEventArgs {IsRunning = true});

                // Now let other endpoints know we're online so they can go look up our subscription information
                Send(new EndpointOnline { SubscriberName = Configuration.CurrentEndpointInputChannel.DestinationName });
            } 
            catch (Exception e) 
            {
                IsRunning = false;
                throw new FatalException("Unable to start ServiceBus.", e);
            }
        }

        /// <summary>
        /// Stops the service bus.
        /// </summary>
        public void Stop()
        {
            Dispose();
        }

        /// <summary>
        /// Sends the specified message. The service bus will first lookup all registered destinations for the message type 
        /// using the <see cref="IDestinationRegistry"/> instance configured.
        /// </summary>
        /// <typeparam name="T">The message type</typeparam>
        /// <param name="message">The message to send.</param>
        public void Send<T>(T message) where T : class
        {
            Send(message, Guid.Empty);
        }

        /// <summary>
        /// Sends the specified message. The service bus will first lookup all registered destinations for the message type 
        /// using the <see cref="IDestinationRegistry"/> isntance configured.
        /// </summary>
        /// <typeparam name="T">The message type</typeparam>
        /// <param name="message">The message to send.</param>
        /// <param name="workflowId">The workflow id of the workflow this message is participating in.</param>
        public void Send<T>(T message, Guid workflowId) where T : class
        {
            if (message == null) throw new ArgumentNullException("message");

            var allDestinations = Configuration.DestinationRegistry.GetIntendedDestinations(typeof(T));
            foreach (var destination in allDestinations)
                Send(message, destination, workflowId);
        }

        /// <summary>
        /// Sends the specified message to the specified MessagingChannel.
        /// </summary>
        /// <typeparam name="T">The type of message to send.</typeparam>
        /// <param name="message">The message to send. If the message is null, an ArgumentNullException is thrown.</param>
        /// <param name="messagingChannel">The MessagingChannel. If the MessagingChannel is null, an ArgumentNullException is thrown.</param>
        public void Send<T>(T message, IMessagingChannel messagingChannel) where T : class
        {
            Send(message, messagingChannel, Guid.Empty);
        }

        /// <summary>
        /// Sends the specified message to the specified MessagingChannel.
        /// </summary>
        /// <typeparam name="T">The type of message to send.</typeparam>
        /// <param name="message">The message to send. If the message is null, an ArgumentNullException is thrown.</param>
        /// <param name="messagingChannel">The MessagingChannel. If the MessagingChannel is null, an ArgumentNullException is thrown.</param>
        /// <param name="workflowId">The workflow id of the workflow this message is participating in.</param>
        public void Send<T>(T message, IMessagingChannel messagingChannel, Guid workflowId) where T : class
        {
            AssertThatServiceBusHasBeenStarted();
            if (message == null) throw new ArgumentNullException("message");
            if (messagingChannel == null) throw new ArgumentNullException("messagingChannel");

            // Create task
            var transmissionTask = new TransmissionTask
                                       {
                                           MessagingChannel = messagingChannel, 
                                           Message = message,
                                           WorkflowInstanceId = workflowId
                                       };

            // Enqueue task with distributor
            if (Log.IsDebugEnabled) Log.DebugFormat("Enqueing transmission task to send message: {0}", message.GetType().FullName);
            _taskDistributor.EnqueueTask(transmissionTask);
        }

        /// <summary>
        /// Sends a message designated as a request. Receivers of request messages should reply to the
        /// MessagingChannel provided to this method. This overload will send the request only to the destinations
        /// provided to this method instead of sending to anyone with an interest in the supplied message type.
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="message">The message.</param>
        /// <param name="destination">The receiving MessagingChannel.</param>
        /// <param name="replyDestination">The reply MessagingChannel.</param>
        /// <returns>
        /// An identifer uniquely identifying the request/reply exchange of message.
        /// </returns>
        public Guid Request<T>(T message, IEnumerable<IMessagingChannel> destination, IMessagingChannel replyDestination) where T : class 
        {
            AssertThatServiceBusHasBeenStarted();
            if (message == null) throw new ArgumentNullException("message");
            if (replyDestination == null) throw new ArgumentNullException("replyDestination");

            if (Log.IsDebugEnabled) Log.DebugFormat("Enqueing transmission task to send message: {0}. Will expect a reply to {1}.", message.GetType().FullName, replyDestination.DestinationName);

            var requestIdentifier = Guid.NewGuid();
            foreach (var d in destination)
            {
                // Enqueue task with distributor
                var task = new TransmissionTask
                {
                    MessagingChannel = d,
                    Message = message,
                    ReplyDestination = replyDestination,
                    RequestIdentifier = requestIdentifier
                };
                _taskDistributor.EnqueueTask(task);
            }

            return requestIdentifier;
        }

        /// <summary>
        /// Sends a message designated as a request. Receivers of request messages should reply to the
        /// MessagingChannel provided to this method. This overload will send the request to anybody registered 
        /// with an interest in this message type.
        /// </summary>
        /// <typeparam name="T">The message type.</typeparam>
        /// <param name="message">The message.</param>
        /// <param name="replyDestination">The reply MessagingChannel.</param>
        /// <returns>An identifer uniquely identifying the request/reply exchange of message.</returns>
        public Guid Request<T>(T message, IMessagingChannel replyDestination) where T : class
        {
            var allDestinations = Configuration.DestinationRegistry.GetIntendedDestinations(typeof(T));

            return Request(message, allDestinations, replyDestination);
        }

        /// <summary>
        /// Sends a message designated as a request. Receivers of request messages should reply to the 
        /// MessagingChannel provided to them. When using this overload, the reply MessagingChannel is the current 
        /// endpoint's input queue. The request is sent to all parties registered with an interest in 
        /// the supplied message type.
        /// </summary>
        /// <typeparam name="T">The message type.</typeparam>
        /// <param name="message">The message.</param>
        /// <returns>An identifer uniquely identifying the request/reply exchange of message.</returns>
        public Guid Request<T>(T message) where T : class
        {
            return Request(message, Configuration.CurrentEndpointInputChannel);
        }

        /// <summary>
        /// Creates a subscription based on the type of message expected
        /// and the expected MessagingChannel. This overload sends a single 
        /// <seealso cref="EndpointSubscriptionChanged"/> message once all subscriptions are created.
        /// </summary>
        /// <typeparam name="T">The message handler type.</typeparam>
        /// <param name="expectedMessagingChannel">The expected MessagingChannel.</param>
        public void Subscribe<T>(IMessagingChannel expectedMessagingChannel) where T : class
        {
            Subscribe<T>(new List<IMessagingChannel> {expectedMessagingChannel});
        }

        /// <summary>
        /// Creates multiple subscriptions based on the type of message expected and provided 
        /// destinations. This overload sends only a single <seealso cref="EndpointSubscriptionChanged"/> 
        /// message once all subscriptions are created.
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="expectedDestinations">The expected destinations.</param>
        public void Subscribe<T>(IEnumerable<IMessagingChannel> expectedDestinations) where T : class 
        {
            Subscribe(typeof(T), expectedDestinations);
        }

        /// <summary>
        /// Creates multiple subscriptions based on the type of message expected and provided
        /// destinations. This overload sends only a single <seealso cref="EndpointSubscriptionChanged"/>
        /// message once all subscriptions are created.
        /// </summary>
        /// <param name="messageType">Type of the message.</param>
        /// <param name="expectedDestinations">The expected destinations.</param>
        public void Subscribe(Type messageType, IEnumerable<IMessagingChannel> expectedDestinations)
        {
            AssertThatServiceBusHasBeenStarted();

            // Make sure we have a message handler for this message discovered already
            AssertThatASuitableMessageHandlerHasBeenRegistered(messageType);

            var destinationMaps = new Dictionary<IMessagingChannel, List<Type>>();
            foreach (var destination in expectedDestinations)
            {
                // Create and save subscription
                Configuration.DestinationRegistry.AppendSubscriptionData(destination, new[] { messageType });
                _taskDistributor.EnqueueTask(new SubscriptionTask(destination));

                // Build up this dictionary so we can use it in CLR event for subscription update
                if (destinationMaps.ContainsKey(destination))
                    destinationMaps[destination].Add(messageType);
                else
                    destinationMaps.Add(destination, new List<Type>(new[] { messageType }));
            }

            // Save subscription data
            Configuration.DestinationRegistry.SaveCurrentSubscriptionData();

            // Alert interested parties via CLR event that subscriptions have changed
            NotifySubscriptionsUpdated(destinationMaps, true);

            // Alert other endpoints that our subscriptions have changed
            Send(new EndpointSubscriptionChanged { EndpointName = Context.ServiceBus.Configuration.CurrentEndpointInputChannel.DestinationName });
        }

        /// <summary>
        /// Creates a new subscription based on the type of message expected. This overload assumes 
        /// the expected MessagingChannel is a PubSubChannel with the same name as the message type and 
        /// cascades into the overload that accepts a single IMessagingChannel.
        /// </summary>
        /// <typeparam name="T">The message type.</typeparam>
        public void Subscribe<T>() where T : class 
        {
            Subscribe<T>(new PubSubChannel(typeof (T).FullName));
        }

        /// <summary>
        /// Deletes subscriptions matching the type of message expected and specified MessagingChannel.
        /// </summary>
        /// <typeparam name="T">The message type.</typeparam>
        /// <param name="expectedMessagingChannel">The expected MessagingChannel.</param>
        public void UnSubscribe<T>(IMessagingChannel expectedMessagingChannel) where T : class
        {
            Configuration.DestinationRegistry.InvalidateSubscription(expectedMessagingChannel, typeof(T));
            Configuration.DestinationRegistry.SaveCurrentSubscriptionData();

            // Alert interested parties via CLR event that subscriptions have changed
            if(SubscriptionsUpdated != null)
            {
                var eventArgs = new SubscriptionUpdateEventArgs();
                eventArgs.SubscriptionsTerminated.Add(expectedMessagingChannel, new[] {typeof (T)});

                SubscriptionsUpdated(this, eventArgs);
            }

            // Alert other endpoints that our subscriptions have changed
            Context.ServiceBus.Send(new EndpointSubscriptionChanged { EndpointName = Context.ServiceBus.Configuration.CurrentEndpointInputChannel.DestinationName });
        }

        /// <summary>
        /// Resets the message sent count and returns the number of messages sent since last time 
        /// this method was called on the instance.
        /// </summary>
        /// <returns></returns>
        public ulong ResetMessageSentCount()
        {
            lock (_msgCountSync)
            {
                var cnt = _sendCount;
                _sendCount = 0;

                return cnt;
            }
        }

        /// <summary>
        /// Resets the message received count and returns the number of messages sent since last time 
        /// this method was called on the instance.
        /// </summary>
        /// <returns></returns>
        public ulong ResetMessageReceivedCount()
        {
            lock (_msgCountSync)
            {
                var cnt = _receiveCount;
                _receiveCount = 0;

                return cnt;
            }
        }

        #endregion

        #region Event Handlers

        private void MessageHandlerRegistry_HandlersRegistered(object sender, HandlersRegisteredEventArgs e)
        {
            // Create subscription tasks for each message-type-to-MessagingChannel mapping
            foreach (var reg in e.Registrations)
            {
                foreach (var destination in reg.ExpectedDestinations)
                {
                    var d = destination;    // Avoid access to modified closure??
                    var kvp = _queuedSubscriptions.SingleOrDefault(p => p.Key == d);
                    if (kvp.Value == null || kvp.Key == null)
                        _queuedSubscriptions.Enqueue(new KeyValuePair<IMessagingChannel, List<Type>>(destination, new List<Type> { reg.MessageType }));
                    else
                        kvp.Value.Add(reg.MessageType);
                }
            }

            // Append subscriber data so we later save subscription data for all other endpoints to discover
            foreach (var kvp in _queuedSubscriptions)
                Configuration.DestinationRegistry.AppendSubscriptionData(kvp.Key, kvp.Value.ToArray());

            // Inform MessagingChannel registry that we're done appending subscription data for our own endpoint subscriptions
            Configuration.DestinationRegistry.SaveCurrentSubscriptionData();

            // Now subscription tasks to task distributor
            // If this instance has not been started yet, store the subscription tasks until it is started.
            if (IsRunning)
                SubmitQueuedSubscriptionRequestsToTaskDistributor();
        }

        /// <summary>
        /// Handles the MessageReceived event of the TaskDistributor instance by in turn raising a MessageReceived event.
        /// </summary>
        /// <param name="sender">The source of the event.</param>
        /// <param name="e">The <see cref="Reactor.Messaging.ServiceBus.MessageEventArgs"/> instance containing the event data.</param>
        private void TaskDistributor_MessageReceived(object sender, MessageEventArgs e)
        {
            lock (_msgCountSync)
            {
                _receiveCount++;
            }

            // Bubble received event upward
            if (MessageReceived != null) MessageReceived(this, e);
        }

        /// <summary>
        /// Handles the MessageSent event of the TaskDistributor instance by in turn raising a MessageSent event.
        /// </summary>
        /// <param name="sender">The source of the event.</param>
        /// <param name="e">The <see cref="Reactor.Messaging.ServiceBus.MessageEventArgs"/> instance containing the event data.</param>
        private void TaskDistributor_MessageSent(object sender, MessageEventArgs e)
        {
            lock (_msgCountSync)
            {
                _sendCount++;
            }

            // Bubble sent event upward
            if (MessageSent != null) MessageSent(this, e);
        }

        #endregion

        #region Private Methods

        private void AssertThatServiceBusHasBeenStarted()
        {
            if (!IsRunning) throw new InvalidOperationException("Cannot begin using the service bus before it has been configured and started.");
        }

        private void DetermineDefaultNumberOfConsumingSessions()
        {
            int defaultNumberOfConsumingSessions;
            var fromConfig = ConfigurationManager.AppSettings["DefaultNumberOfConsumingSessions"];
            int.TryParse(fromConfig, out defaultNumberOfConsumingSessions);
            _defaultNumberOfConsumingSessions = (defaultNumberOfConsumingSessions > 0) ? defaultNumberOfConsumingSessions : 3;
        }

        private void AssertThatASuitableMessageHandlerHasBeenRegistered(Type msgType)
        {
            var handlers = Configuration.MessageHandlerRegistry.GetHandlerTypesForMessageType(msgType);
            if (handlers.Count == 0)
                throw new InvalidOperationException("Unable to create subscription because no suitable message handler to take delivery of the message type was discovered.");
        }

        private void SubmitQueuedSubscriptionRequestsToTaskDistributor()
        {
            if (_queuedSubscriptions.Count <= 0) return;

            var destinationMaps = new Dictionary<IMessagingChannel, List<Type>>(_queuedSubscriptions.Count);
            var cnt = _queuedSubscriptions.Count;
            for (var i = 0; i < cnt; i++)
            {
                KeyValuePair<IMessagingChannel, List<Type>> kvp;
                if (_queuedSubscriptions.TryDequeue(out kvp))
                {
                    _taskDistributor.EnqueueTask(new SubscriptionTask(kvp.Key));

                    if (destinationMaps.ContainsKey(kvp.Key))
                        destinationMaps[kvp.Key] = kvp.Value;
                    else
                        destinationMaps.Add(kvp.Key, kvp.Value);
                }
            }

            NotifySubscriptionsUpdated(destinationMaps, true);
        }

        private void NotifySubscriptionsUpdated(Dictionary<IMessagingChannel, List<Type>> destinationMaps, bool isCreationEvent)
        {
            // Alert interested parties via CLR event that subscriptions have changed
            if (SubscriptionsUpdated == null) return;

            var eventArgs = new SubscriptionUpdateEventArgs();
            foreach (var kvp in destinationMaps)
            {
                if (isCreationEvent)
                    eventArgs.SubscriptionsCreated.Add(kvp.Key, kvp.Value.ToArray());
                else
                    eventArgs.SubscriptionsTerminated.Add(kvp.Key, kvp.Value.ToArray());
            }

            SubscriptionsUpdated(this, eventArgs);
        }

        #endregion

        #region Implementation of IDisposable

        /// <summary>
        /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
        /// </summary>
        /// <filterpriority>2</filterpriority>
        protected override void Dispose(bool disposing)
        {
            // Tear down message handler registry
            if (Configuration.MessageHandlerRegistry != null)
                Configuration.MessageHandlerRegistry.HandlersRegistered -= MessageHandlerRegistry_HandlersRegistered;

            // Terminate no durable subscriptions if needed and dispose MessagingChannel registry
            if (Configuration.DestinationRegistry != null)
            {
                if(Configuration.TerminateNonDurableSubscriptionsOnShutdown)
                    Configuration.DestinationRegistry.InvalidateSubscriptionsOnNonDurableDestinations();

                Configuration.DestinationRegistry.Dispose();
            }

            // Shut down transport gateway
            if(_nmsGateway != null)
                _nmsGateway.Shutdown();

            // Tear down task distributor
            if(_taskDistributor != null)
            {
                _taskDistributor.Stop();
                _taskDistributor.MessageReceived -= TaskDistributor_MessageReceived;
                _taskDistributor.MessageSent -= TaskDistributor_MessageSent;
                _taskDistributor.Dispose();
            }

            IsRunning = false;

            if(RunStatusChanged != null)
                RunStatusChanged(this, new RunStatusChangedEventArgs {IsRunning = false});

            base.Dispose(disposing);
        }

        #endregion
    }
}
